package top.caozj.task;

import com.hivemq.client.mqtt.MqttClient;
import com.hivemq.client.mqtt.mqtt3.Mqtt3AsyncClient;
import com.hivemq.client.mqtt.mqtt3.message.publish.Mqtt3Publish;
import lombok.extern.log4j.Log4j2;
import top.caozj.conf.AppConfig;
import top.caozj.conf.DataQuery;
import top.caozj.conf.Datasource;
import top.caozj.pojo.RawData;
import top.caozj.proto.MonitorDataProtocol;
import top.caozj.util.DataSourceType;
import com.hivemq.client.mqtt.mqtt3.Mqtt3BlockingClient;

import java.io.UnsupportedEncodingException;
import java.nio.charset.StandardCharsets;
import java.sql.*;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.Instant;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Date;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;


/**
 * Copyright (c) 2010, 2025, xjk All rights reserved.
 * 数据上床任务类
 * @author caozj thucaozj@outlook.com
 * @version 1.0
 * @date 2025-03-26
 * @since 20
 */
@Log4j2
public class MqttUpload {

    //mqtt客户端
    Mqtt3AsyncClient client;
//    static ExecutorService executorService = Executors.newFixedThreadPool(10);


    public MqttUpload() throws UnsupportedEncodingException {
        client = MqttClient.builder()
                .useMqttVersion3()
                .identifier("auto-script")
                .serverHost(AppConfig.mqttConf.getHost())
                .serverPort(AppConfig.mqttConf.getPort()) // 默认 MQTT 端口
                .simpleAuth()
            .username(AppConfig.mqttConf.getUsername())
                .password(AppConfig.mqttConf.getPassword().getBytes(StandardCharsets.UTF_8))
                .applySimpleAuth()
                .buildAsync();

        // 连接到 MQTT 服务端
        try {
            client.connect();
            log.info("Connected to MQTT broker successfully.");
        } catch (Exception e) {
            log.error("Failed to connect to MQTT broker: {}", e.getMessage());

            System.exit(-1);
        }
        //
    }

    /**
     * 处理数据库查询的原始数据
     * @param data 原始数据
     */
    public void processRawData(RawData data){
        String sysId = AppConfig.sensorCodeMap.get(data.getId());

        try {
            float floatValue =Float.parseFloat(data.getValue());

            java.util.Date date = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(data.getTime());

            publishData(MonitorDataProtocol.MonitoryType.valueOf(AppConfig.sensorType.get(data.getId())), sysId, floatValue, date);
        }catch (Exception e){
            log.error("processRawData error");
//            e.printStackTrace();
        }

    }

    public  void  publishData(MonitorDataProtocol.MonitoryType type, String sensorNo, float value, Date receiveTime){

        log.debug("type:{}  sensorNo:{} value:{} receiveTime: {}" ,type ,sensorNo, value , receiveTime);
        MonitorDataProtocol.ComplexData.Builder complexData = MonitorDataProtocol.ComplexData.newBuilder();
        MonitorDataProtocol.SensorData.Builder sensorData = MonitorDataProtocol.SensorData.newBuilder();
        sensorData.setMonitorType(type);
        sensorData.setSensorNo(sensorNo);
        sensorData.setUpTime(receiveTime.getTime());

        switch (type) {
            case RHS:
                    sensorData.setRhs(MonitorDataProtocol.RHSRealTime.newBuilder().addTemperature(value));
                    break;
            case WDS :
                sensorData.setWds(MonitorDataProtocol.WDSRealTime.newBuilder().addWindSpeed(value));
                break;
            case WDD : sensorData.setWdd(MonitorDataProtocol.WDDRealTime.newBuilder().addWindDirection(value));
                break;
            case PWS : sensorData.setPws(MonitorDataProtocol.PWSRealTime.newBuilder().addRainfall(value));
                break;
            case PRS : sensorData.setPrs(MonitorDataProtocol.PRSRealTime.newBuilder().addPressure(value));
                break;
            case RAD : sensorData.setRad(MonitorDataProtocol.RADRealTime.newBuilder().addRadiation(value));
                break;
            case ILL : sensorData.setIll(MonitorDataProtocol.ILLRealTime.newBuilder().addIllumination(value));
                break;
            case UVT : sensorData.setUvt(MonitorDataProtocol.UVTRealTime.newBuilder().addUltraviolet(value));
                break;
            case RSG : sensorData.setRsg(MonitorDataProtocol.RSGRealTime.newBuilder().addStrain(value));
                break;
            case LMZ : sensorData.setLmz(MonitorDataProtocol.LMZRealTime.newBuilder().addTemperature(value));
                break;
            default :
                log.error("No Such Type:" + type);

        }
        complexData.addSensorData(sensorData);
        byte[] a = complexData.build().toByteArray();
//        mqtt.publish(0, false, "b/320001/rt", a);

        Mqtt3Publish publish = Mqtt3Publish.builder()
                .topic("r/r6540000001/rt")
                .payload(a)
                .build();

//       try {
//            synchronized (client) {
//                client.publish(publish);
//            }
//        } catch (Exception e) {
//            log.error("mqtt publish error");
//        }

        client.publish(publish);
    }

    /**
     * 处理离线数据
     * @param id

     */
    public void processOffSetData(String id, String tTime, Double value) {
        try{
        String sysId = AppConfig.sensorCodeMap.get(id);


        java.util.Date date = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(tTime);

        float floatValue = value.floatValue();
        publishData(MonitorDataProtocol.MonitoryType.valueOf(AppConfig.sensorType.get(id)), sysId, floatValue, date);

        }catch (Exception e){
            log.error("processRawData error");
            //            e.printStackTrace();
        }

    }
}
